

package org.example.flinksql.test;

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.StatementSet;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.example.flinksql.test.custom.TransEidFunction;

/**
 * 分表数据聚合、转换demo，读取和写入均为JDBC Connector
 * 源表有8张分表，t_enterprise0~7
 * 目标表为t_enterprise
 * 转换逻辑为：每个eid前拼接”eid“字符串
 */
public class AggregateAndTransMultTable {

	public static void main(String[] args) throws Exception {
		// set up the streaming execution environment
		final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
		EnvironmentSettings bsSettings = EnvironmentSettings.newInstance().useBlinkPlanner().inStreamingMode().build();
		StreamTableEnvironment tbEnv = StreamTableEnvironment.create(env, bsSettings);

		for(int i = 0; i < 8; i ++) {
			tbEnv.executeSql("CREATE TEMPORARY TABLE t_enterprise_" + i +" (" +
					"eid STRING," +
					"name STRING," +
					"credit_no STRING," +
					"category_new INT," +
					"regist_capi INT," +
					"actual_capi INT," +
					"status INT) WITH (" +
					"'connector' = 'jdbc'," +
					"'url' = 'jdbc:mysql://under.azure:3306/flink-test?useSSL=false'," +
					"'table-name' = 't_enterprise_" + i + "'," +
					"'username' = 'root'," +
					"'password' = '123456'," +
					"'sink.buffer-flush.max-rows' = '0'," +
					"'sink.buffer-flush.interval' = '0'," +
					"'sink.max-retries' = '3'" +
					")");
		}

		// TODO:Sink必须设置主键，当主键冲突时会自动更新数据，如果不设置程序会报错
		tbEnv.executeSql("CREATE TEMPORARY TABLE t_enterprise (" +
				"eid STRING," +
				"name STRING," +
				"credit_no STRING," +
				"category_new INT," +
				"regist_capi INT," +
				"actual_capi INT," +
				"status INT," +
				"PRIMARY KEY (eid) NOT ENFORCED) WITH (" +
				"'connector' = 'jdbc'," +
				"'url' = 'jdbc:mysql://under.azure:3306/flink-test?useSSL=false'," +
				"'table-name' = 't_enterprise'," +
				"'username' = 'root'," +
				"'password' = '123456'," +
				"'sink.buffer-flush.max-rows' = '0'," +
				"'sink.buffer-flush.interval' = '0'," +
				"'sink.max-retries' = '3'" +
				")");

		tbEnv.createFunction("transeid", TransEidFunction.class);

		//test transfunc
//		tbEnv.sqlQuery("select transfunc(eid,name,credit_no,category_new," +
//				"regist_capi,actual_capi,status) from t_enterprise_0")
//				.execute()
//				.print();


		// 使用StatementSet批量新增insert任务
		StatementSet statementSet = tbEnv.createStatementSet();
		for (int i = 0; i < 8; i++) {
			statementSet.addInsertSql("insert into t_enterprise " +
					"select transeid(eid) as eid,name,credit_no,category_new," +
					"regist_capi,actual_capi,status from t_enterprise_" + i);
		}
		statementSet.execute();


		// execute program
//		env.execute("Flink Streaming Java API Skeleton");
	}
}
